package com.gjy.kafka.resolve.config;

import com.fasterxml.jackson.databind.deser.std.StringDeserializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * 消费者通用配置
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 15:19:26
 */
public class KafkaConsumerConfig {

    /**
     * @param groupId 消费组 ID
     */
    public static Properties buildConfig(String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 关闭自动提交，改为手动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 读取已提交的事务消息
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

        // 业务调优
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10_000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3_000);
        // 必须指定消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return props;
    }

}
